package com.retail.order.consumer;

import com.alibaba.fastjson.JSON;
import com.rabbitmq.client.Channel;
import com.retail.common.core.domain.Result;
import com.retail.order.config.ExecutorConfig;
import com.retail.order.config.RabbitMQConfig;
import com.retail.order.service.OrderSkuService;
import com.retail.order.vo.OrderRequest;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.annotation.RabbitHandler;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.stereotype.Component;

import java.io.IOException;
import java.util.concurrent.Executor;
import java.util.concurrent.TimeUnit;

/**
 * @Author: 靳家振
 * @Date: 2024/3/6 10:24
 * @Description:
 **/

@Component
@RabbitListener(queues = RabbitMQConfig.INSERT_QUEUE)
public class Consumer {
    @Autowired
    private StringRedisTemplate redisTemplate;
    @Autowired
    private OrderSkuService orderSkuService;
    @Autowired
    private ExecutorConfig executorConfig;

    private final String response ="RESPONSE";

    @RabbitHandler
    public void take(OrderRequest request, Channel channel, Message message) throws IOException {
        String messageId = message.getMessageProperties ().getMessageId ();
        long deliveryTag = message.getMessageProperties ().getDeliveryTag ();
        if (Boolean.FALSE.equals (redisTemplate.hasKey ("INSERT_ORDER_RD_" + messageId))){
            redisTemplate.opsForValue ().set ("INSERT_ORDER_RD_"+messageId,deliveryTag+"",30, TimeUnit.SECONDS);
        }
        Long size = redisTemplate.opsForSet ().size ("INSERT_ORDER_CF_" + messageId);
        try {
            if (size==0){
                Executor executor = executorConfig.insertOrderExecutors ();
                executor.execute (()->{
                    OrderRequest orderResponse = orderSkuService.getOrderId (request);
                    redisTemplate.opsForValue ().set (orderResponse.getUuid ()+response, JSON.toJSONString (orderResponse),5,TimeUnit.MINUTES);
                });

                redisTemplate.opsForSet ().add ("INSERT_ORDER_CF_"+messageId,messageId);
                redisTemplate.expire ("INSERT_ORDER_CF_"+messageId,30,TimeUnit.SECONDS);
                channel.basicAck (deliveryTag,false);
            }else {
                System.out.println ("消息重复了");
                channel.basicReject (deliveryTag,false);
            }
        } catch (IOException e) {
            System.out.println ("消息重新入队");
            String s = redisTemplate.opsForValue ().get ("INSERT_ORDER_RD_" + messageId);
            long l = Long.parseLong (s);
            if (deliveryTag >= (l + 2)) {
                System.out.println ("确实消费不了了");
                channel.basicNack (deliveryTag,false,false);
            }else {
                System.out.println ("消息重复入队");
                channel.basicNack (deliveryTag,false,true);
            }
        }
    }

}
